package com.bangcommunity.bbframe.springredis.rwclient;

import com.bangcommunity.bbframe.common.exception.ValidateUtil;
import com.bangcommunity.bbframe.common.utils.lang.CollectionUtils;
import com.bangcommunity.bbframe.common.utils.lang.StringUtils;
import com.google.common.collect.Maps;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import redis.clients.jedis.JedisShardInfo;
import redis.clients.jedis.ShardedJedis;

import java.io.ByteArrayInputStream;
import java.io.InputStream;
import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.locks.ReentrantLock;

public class RedisRWClient {

    private static Logger logger = LoggerFactory.getLogger(RedisRWClient.class);

    private List<JedisShardInfo> clients = null;
    private List<JedisShardInfo> master = new ArrayList<JedisShardInfo>();
    private List<JedisShardInfo> slave = new ArrayList<JedisShardInfo>();
    private final static String test_flag = "_t_";
    private int count = 0;

    private RedisRWClient(InputStream conf) {
        init(conf);
    }

    /**
     * redis链接库 uri
     * =http://password:aos12321fadfafd293rrf32r9fija89uf230r9faofj2309u8302fj@139.217.10.24:7001/1
     *
     * password: autho值 /1 为使用的db
     *
     * 该项目使用1库
     *
     * http://password:aos12321fadfafd293rrf32r9fija89uf230r9faofj2309u8302fj@139.217.10.24:56379/1
     * http://password:aos12321fadfafd293rrf32r9fija89uf230r9faofj2309u8302fj@139.217.10.24:56380/1
     * http://password:aos12321fadfafd293rrf32r9fija89uf230r9faofj2309u8302fj@139.217.10.24:56381/1
     */
    private void init(InputStream conf) {

        Properties prop = new Properties();
        try {
            prop.load(conf);
            JedisShardInfo shardInfo = null;
            clients = new ArrayList<>();
            for (Object key : prop.keySet()) {
                String val = (String) prop.get(key);
                shardInfo = new JedisShardInfo(val);
                clients.add(shardInfo);
            }
            logger.info("初始化redis服务成功");
        } catch (Exception e) {
            logger.error("初始化redis服务失败", e);
        }
    }

    public static RedisRWClient getInstance(String key, InputStream conf) {
        ValidateUtil.checkNotEmpty(key, "redis标识不能为空");
        return RedisRWClientHolder.createIfNotExist(key, conf);
    }

    private static class RedisRWClientHolder {
        private static ReentrantLock lock = new ReentrantLock();
        private static ConcurrentMap<String, RedisRWClient> clientMap = Maps.newConcurrentMap();

        private static RedisRWClient createIfNotExist(String key, InputStream conf) {
            RedisRWClient redisRWClient = clientMap.get(key);
            if (null != redisRWClient) {
                return redisRWClient;
            }
            if (null != conf) {
                logger.info("init redis rw client for {}", key);
                redisRWClient = new RedisRWClient(conf);
                clientMap.put(key, redisRWClient);
            } else {
                logger.warn("not get rwredisclient, no redis conf for {}", key);
            }
            return redisRWClient;
        }
    }

    /**
     * 分配主从节点
     */
    public boolean getMasterSlave() {

        int masterIndex = 0;
        logger.warn("redis>>>>>>主从切换");
        // 轮询获取主redis配置
        for (int i = 0; i < this.clients.size(); i++) {
            JedisShardInfo jedisShardInfo = this.clients.get(i);
            List<JedisShardInfo> masterTest = Arrays.asList(jedisShardInfo);
            String host = jedisShardInfo.getHost();
            int port = jedisShardInfo.getPort();
            ShardedJedis sharding = new ShardedJedis(masterTest);
            String result = null;
            try {
                result = sharding.set(test_flag, test_flag);
            } catch (Exception e) {
                logger.warn(">>>>>>>>>>" + host + ":" + port + "作为Redis主机失败:" + e.getMessage());
                disconnect(sharding);
                continue;
            }
            sharding.close();
            if ("OK".equals(result)) {
                this.master = Arrays.asList(jedisShardInfo);
                logger.info(">>>>>>>>>>" + host + ":" + port + "作为Redis主机成功");
                masterIndex = i;
                break;
            }
        }
        // 轮询获取从redis配置
        this.slave = new ArrayList<>();
        for (int j = 0; j < this.clients.size(); j++) {
            if (masterIndex == j) {
                continue;
            }
            JedisShardInfo jedisShardInfo = this.clients.get(j);
            List<JedisShardInfo> test = Arrays.asList(jedisShardInfo);
            ShardedJedis sharding = new ShardedJedis(test);
            String host = jedisShardInfo.getHost();
            int port = jedisShardInfo.getPort();
            String result = null;
            try {
                result = sharding.get(test_flag);
            } catch (Exception e) {
                logger.warn(">>>>>>>>>>" + host + ":" + port + "作为Redis从机失败:" + e.getMessage());
                disconnect(sharding);
                continue;
            }
            sharding.close();
            if (null != result) {
                this.slave.add(jedisShardInfo);
                logger.info(">>>>>>>>>>" + host + ":" + port + "作为Redis从机成功");
            }
        }
        if (slave.size() == 0 && master.size() != 0) {
            slave.add(master.get(0));
        }
        return CollectionUtils.isNotEmpty(master) && CollectionUtils.isNotEmpty(slave);
    }

    public void disconnect(ShardedJedis sharding) {
        try {

            sharding.close();
        } catch (Exception e) {
            e.getMessage();
        }
    }

    /**
     * 设置超时时间put字符串方法
     * 
     * @param key
     * @param expires
     * @param value
     */
    public void putex(String key, int expires, String value) {
        ShardedJedis sharding = new ShardedJedis(master);
        try {
            String result = sharding.setex(key, expires, value);
            sharding.close();
        } catch (Exception e) {
            logger.error("putex>>>>>存储key=" + key + "失败:" + e.getMessage());
            disconnect(sharding);
            // 每次处理主从转换次数1次，超过一次不再处理
            if (count < 1) {
                count++;
                getMasterSlave();
                putex(key, expires, value);
            }
        } finally {
            count = 0;
        }
    }

    public boolean setnx(String key, String value) {
        ShardedJedis sharding = new ShardedJedis(master);
        try {
            Long result = sharding.setnx(key, value);
            sharding.close();
            return result == 1;
        } catch (Exception e) {
            logger.error("putex>>>>>存储key=" + key + "失败:" + e.getMessage());
            disconnect(sharding);
            // 每次处理主从转换次数1次，超过一次不再处理
            if (count < 1) {
                count++;
                getMasterSlave();
                return setnx(key, value);
            }
        } finally {
            count = 0;
        }
        return false;
    }

    public boolean expire(String key, int seconds) {
        ShardedJedis sharding = new ShardedJedis(master);
        try {
            Long result = sharding.expire(key, seconds);
            sharding.close();
            return result == 1;
        } catch (Exception e) {
            logger.error("putex>>>>>存储key=" + key + "失败:" + e.getMessage());
            disconnect(sharding);
            // 每次处理主从转换次数1次，超过一次不再处理
            if (count < 1) {
                count++;
                getMasterSlave();
                return expire(key, seconds);
            }
        } finally {
            count = 0;
        }
        return false;
    }

    public void removeKey(String key) {
        ShardedJedis sharding = new ShardedJedis(master);
        try {
            sharding.del(key);
            sharding.close();
        } catch (Exception e) {
            logger.error("del>>>>>存储key=" + key + "失败:" + e.getMessage());
            disconnect(sharding);
            // 每次处理主从转换次数1次，超过一次不再处理
            if (count < 1) {
                count++;
                getMasterSlave();
                removeKey(key);
            }
        } finally {
            count = 0;
        }
    }

    /**
     * 计数器
     * 
     * @param key
     *            计数器key值
     * @param expires
     *            存活时间
     * @return
     */
    public long increx(String key, int expires) {
        ShardedJedis sharding = new ShardedJedis(master);
        long in = 0L;
        try {
            in = sharding.incr(key);
            if (in == 1L) {
                sharding.expire(key, expires);
            }
            sharding.close();
        } catch (Exception e) {
            logger.error("increx>>>>>存储key=" + key + "失败:" + e.getMessage());
            disconnect(sharding);
            // 每次处理主从转换次数1次，超过一次不再处理
            if (count < 1) {
                count++;
                getMasterSlave();
                return increx(key, expires);
            }
        } finally {
            count = 0;
        }
        return in;
    }

    public long increx(String key) {
        ShardedJedis sharding = new ShardedJedis(master);
        long in = 0L;
        try {
            in = sharding.incr(key);
            sharding.close();
        } catch (Exception e) {
            logger.error("increx>>>>>存储key=" + key + "失败:" + e.getMessage());
            disconnect(sharding);
            // 每次处理主从转换次数1次，超过一次不再处理
            if (count < 1) {
                count++;
                getMasterSlave();
                return increx(key);
            }
        } finally {
            count = 0;
        }
        return in;
    }

    public long decr(String key) {
        ShardedJedis sharding = new ShardedJedis(master);
        long in = 0L;
        try {
            in = sharding.decr(key);
            sharding.close();
        } catch (Exception e) {
            logger.error("increx>>>>>存储key=" + key + "失败:" + e.getMessage());
            disconnect(sharding);
            // 每次处理主从转换次数1次，超过一次不再处理
            if (count < 1) {
                count++;
                getMasterSlave();
                return decr(key);
            }
        } finally {
            count = 0;
        }
        return in;
    }

    /**
     * put字符串方法
     * 
     * @param key
     * @param value
     */
    public void put(String key, String value) {
        ShardedJedis sharding = new ShardedJedis(master);
        try {
            String result = sharding.set(key, value);
            sharding.close();
        } catch (Exception e) {
            e.printStackTrace();
            logger.error("put>>>>>存储key=" + key + "失败:" + e.getMessage());
            disconnect(sharding);
            // 每次处理主从转换次数1次，超过一次不再处理
            if (count < 1) {
                count++;
                getMasterSlave();
                put(key, value);
            }
        } finally {
            count = 0;
        }
    }

    /**
     * 设置超时时间put Map<String, String>方法
     * 
     * @param key
     * @param expire
     * @param value
     */
    public void putMapex(String key, int expire, Map<String, String> value) {
        ShardedJedis sharding = new ShardedJedis(master);
        String result = "";
        try {
            result = sharding.hmset(key, value);
            sharding.expire(key, expire);
            sharding.close();
        } catch (Exception e) {
            logger.error("putMapex>>>>>存储key=" + key + "失败:" + e.getMessage());
            disconnect(sharding);
            // 每次处理主从转换次数1次，超过一次不再处理
            if (count < 1) {
                count++;
                getMasterSlave();
                putMapex(key, expire, value);
            }
        } finally {
            count = 0;
        }
    }

    /**
     * put Map<String, String>方法
     * 
     * @param key
     * @param value
     */
    public void putMap(String key, Map<String, String> value) {
        ShardedJedis sharding = new ShardedJedis(master);
        String result = "";
        try {
            result = sharding.hmset(key, value);
            sharding.close();
        } catch (Exception e) {
            logger.error("putMap>>>>>存储key=" + key + "失败:" + e.getMessage());
            disconnect(sharding);
            // 每次处理主从转换次数1次，超过一次不再处理
            if (count < 1) {
                count++;
                getMasterSlave();
                putMap(key, value);
            }
        } finally {
            count = 0;
        }
    }

    /**
     * 获取Map<String,String> value方法
     * 
     * @param key
     * @return
     */
    public Map<String, String> getMap(String key) {
        ShardedJedis sharding = new ShardedJedis(slave);
        try {
            Map<String, String> map = sharding.hgetAll(key);
            sharding.close();
            return map;
        } catch (Exception e) {
            logger.error(">>>>>获取key=" + key + "失败:" + e.getMessage());
            disconnect(sharding);
            // 每次处理主从转换次数1次，超过一次不再处理
            if (count < 1) {
                count++;
                getMasterSlave();
                return getMap(key);
            }
        } finally {
            count = 0;
        }
        return null;
    }

    /**
     * 获取字符串value方法
     * 
     * @param key
     */
    public String get(String key) {
        ShardedJedis sharding = new ShardedJedis(slave);
        try {
            String result = sharding.get(key);
            sharding.close();
            return result;
        } catch (Exception e) {
            logger.error(">>>>>获取key=" + key + "失败:" + e.getMessage());
            disconnect(sharding);
            // 每次处理主从转换次数1次，超过一次不再处理
            if (count < 1) {
                count++;
                getMasterSlave();
                return get(key);
            }
        } finally {
            count = 0;
        }
        return null;
    }

    public boolean isExitsKey(String key) {
        ShardedJedis sharding = new ShardedJedis(slave);
        try {
            boolean flage = sharding.exists(key);
            sharding.close();
            return flage;
        } catch (Exception e) {
            logger.error(">>>>>判断key=" + key + "失败:" + e.getMessage(), e);
            disconnect(sharding);
            // 每次处理主从转换次数1次，超过一次不再处理
            if (count < 1) {
                count++;
                getMasterSlave();
                return isExitsKey(key);
            }

        } finally {
            count = 0;
        }
        return false;
    }

    /**
     * 保存key value
     * 
     * @param key
     * @param value
     *            使用set 集合进行存储(不允许重复数据)
     * @return
     */
    public void sadd(String key, String value) {
        ShardedJedis sharding = new ShardedJedis(master);
        try {
            sharding.sadd(key, value);
            sharding.close();
        } catch (Exception e) {
            logger.error(">>>>>判断addSet=" + key + "失败:" + e.getMessage(), e);
            disconnect(sharding);
            // 每次处理主从转换次数1次，超过一次不再处理
            if (count < 1) {
                count++;
                getMasterSlave();
            }
        } finally {
            count = 0;
        }
    }

    /**
     * 获取 key值的set集合中元素数量
     * 
     * @param key
     * @return
     */
    public long getSetSize(String key) {
        ShardedJedis sharding = new ShardedJedis(slave);
        try {
            long l = sharding.scard(key);
            sharding.close();
            return l;
        } catch (Exception e) {
            logger.error(">>>>>判断key=" + key + "失败:" + e.getMessage(), e);
            disconnect(sharding);
            // 每次处理主从转换次数1次，超过一次不再处理
            if (count < 1) {
                count++;
                getMasterSlave();
                return getSetSize(key);
            }
        } finally {
            count = 0;
        }
        return 0;

    }

    /**
     * redis全局锁,一次获取不到等待1ms继续获取,最长等待maxWaitMs
     * 
     * @param lock
     *            锁标示
     * @param expireSec
     *            锁定时间,秒
     * @param maxWaitMs
     *            获取锁最长等待的时间,毫秒,超过这个时间则获取锁失败
     * @return
     */
    public boolean acquireLock(final String lock, final int expireSec, final long maxWaitMs) {
        logger.debug("acquireLock start:[lock={}][expireSec={}][maxWaitMs={}]", lock, expireSec, maxWaitMs);
        if (StringUtils.isBlank(lock)) {
            return false;
        }
        final String newlock = getLockKey(lock);
        long expireAt = System.currentTimeMillis() + maxWaitMs;
        boolean success = false;
        // 循环判断锁是否一直存在
        while (!success) {
            if (maxWaitMs > 0 && System.currentTimeMillis() > expireAt) {
                break;
            }
            success = this.setnx(newlock, lock);
        }
        this.expire(newlock, expireSec);
        return success;
    }

    private String getLockKey(String lock) {
        return "lock:" + lock;
    }

    /**
     * 解锁
     */
    public void releaseLock(final String lock) {
        logger.debug("releaseLock start:[lock={}]", lock);
        final String newlock = getLockKey(lock);
        this.removeKey(newlock);
    }

    /**
     * 获取全局锁。一次获取不到则失败
     */
    public boolean acquireLockForOnce(final String lock, final int expireSec) {
        logger.debug("acquireLockForOnce start:[lock={}][expireSec={}]", lock, expireSec);
        if (StringUtils.isBlank(lock)) {
            return false;
        }
        final String newlock = getLockKey(lock);
        boolean result = this.setnx(newlock, lock);
        if (result) {
            this.expire(newlock, expireSec);
        }
        return result;

    }

    public static void main(String[] args) throws URISyntaxException, InterruptedException {
        String redisConf = "r1=http://localhost:6379/1";
        RedisRWClient client = RedisRWClient.getInstance("test1", new ByteArrayInputStream(redisConf.getBytes()));
        System.out.println(client.acquireLock("tt", 3, 10000));
        System.out.println(client.acquireLockForOnce("tt", 3));
        System.out.println(client.acquireLock("tt", 3, 10000));
        Thread.sleep(5000);
        System.out.println(client.acquireLock("tt", 3, 10000));
        // redis.getMasterSlave();
        // redis.put("123","234");
        // System.out.println("-----"+);
        // System.out.println("-------"+redis.get("123"));
        // System.out.println("--n2---"+redis.count);

        // http://www.java2s.com:8080/yourpath/fileName.htm?stove=10&path=32&id=4#harvic
        /*
         * URI uri = new URI(
         * "http://password:aos12321fadfafd293rrf32r9fija89uf230r9faofj2309u8302fj@139.217.10.24:7001/1"
         * );
         * 
         * System.out.println("host----"+uri.getHost());
         * System.out.println("port----"+uri.getPort());
         * System.out.println("userinfo----"+uri.getUserInfo());
         * System.out.println("path----"+uri.getPath());
         * System.out.println("password---"+JedisURIHelper.getPassword(uri));
         * 
         * System.out.println("db---"+JedisURIHelper.getDBIndex(uri));
         * 
         * JedisShardInfo jedisShardInfo = new JedisShardInfo(uri);
         */

    }

}
